package com.dsk.connector.sink.common.listener;

import cn.hutool.json.JSONUtil;
import cn.sanenen.sunutils.queue.SMQ;
import com.dsk.connector.commons.trans.TransMsg;
import com.dsk.connector.commons.utils.SerializeUtils;
import com.dsk.connector.sink.common.consumer.SinkConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 消息监听器
 */
public class MessageListener implements Runnable {
    Logger logger= LoggerFactory.getLogger(MessageListener.class);
    Logger sinkPoint=LoggerFactory.getLogger("sinkPoint");
    /**
     * 消息主题
     * 一般一个表对应一个topic
     * 但是如果表太多，就会造成线程超多，给系统带来压力
     * 因此就需要一个线程消费 多个topic
     */
    List<String> topics=new ArrayList<>();

    SinkConsumer consumer=null;

    private boolean isException=false;

    private int exceptionCount=0;

    public MessageListener(){
    }

    /**
     * 构造函数
     * @param topic
     */
    public MessageListener(String... topic){
        for (String t:topic){
            topics.add(t);
        }
    }

    public MessageListener(SinkConsumer consumer,String... topic){
        this.consumer=consumer;
        for (String t:topic){
            topics.add(t);
        }
    }

    public void setConsumer(SinkConsumer consumer){
        this.consumer=consumer;
    }

    public SinkConsumer getConsumer() {
        return consumer;
    }

    public void addTopic(String topic){
        this.topics.add(topic);
    }

    public void removeTopic(String topic){
        this.topics.remove(topic);
    }

    public void consumMessage(String message) throws Exception {
        TransMsg transMsg= SerializeUtils.hessianSerializeToObject(message);
        this.sinkPoint.info("size:"+(transMsg.getDatas()==null?0:transMsg.getDatas().size())+",sinkPoint:"+transMsg.toJsonSourcePoint());
        if(this.isException){
            String jsonMsg=JSONUtil.toJsonStr(transMsg);
            logger.error("发生第"+this.exceptionCount+"次异常的数据是："+jsonMsg);
        }
        this.consumer.consume(transMsg);
    }

    @Override
    public void run() {
        String msg=null;
        String runTopic=null;
        do{
            try{
                if(isException){
                    logger.error("exception occorred in "+runTopic+", "+this.exceptionCount+" times, will retry");
                    if(this.exceptionCount<5 && msg!=null){
                        Thread.sleep(6000);
                        this.consumMessage(msg);
                        msg=null;
                        this.exceptionCount=0;
                        this.isException=false;
                    }else{
                        if(this.exceptionCount>=5){
                            logger.error("exception occorred in "+runTopic+", "+this.exceptionCount+" times, stop consume message");
                            //停止消费,终止运行
                            return ;
                        }else{
                            this.exceptionCount=0;
                            this.isException=false;
                            msg=null;
                        }

                    }

                    continue;
                }
                boolean getMsg=false;
                for(String topic:this.topics){
                    msg=SMQ.pop(topic);
                    if(msg!=null){
                        runTopic=topic;
                        getMsg=true;
                        this.consumMessage(msg);
                        msg=null;
                        this.isException=false;
                    }
                }
                if(!getMsg){
                    //休眠60秒
                    Thread.sleep(60000);
                }
            }catch(Throwable ex){
                //为了避免消息丢失，重新消费失败的消息，注意发生异常后，不能将消息重新放回队列末尾，否则可能会导致新增的数据被旧数据覆盖。
                ex.printStackTrace();
                logger.error(runTopic+"-处理消息失败",ex);
                isException=true;
                this.exceptionCount++;
            }

        }while(1==1);

    }
}
